package rpc

import (
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"reflect"
	"strings"
	"sync"
	"time"

	"zt/go-study/rpc/codec"
)

const MagicNumber = 0x3bef5c

const (
	connected        = "200 Connected to RPC"
	defaultRPCPath   = "/_geerpc_"
	defaultDebugPath = "/debug/geerpc"
)

// 客户端与服务端的通信需要协商一些内容，例如 HTTP 报文，分为 header 和 body 2 部分
// body 的格式和长度通过 header 中的 Content-Type 和 Content-Length 指定，服务端通过解析 header 就能够知道如何从 body 中读取需要的信息。
// 对于 RPC 协议来说，这部分协商是需要自主设计的。
// 为了提升性能，一般在报文的最开始会规划固定的字节，来协商相关的信息。
// 比如第1个字节用来表示序列化方式，第2个字节表示压缩方式，第3-6字节表示 header 的长度，7-10 字节表示 body 的长度。
// 一般来说，涉及协议协商的这部分信息，需要设计固定的字节来传输的。
// 但是为了实现上更简单，GeeRPC 客户端固定采用 JSON 编码 Option，后续的 header 和 body 的编码方式由 Option 中的 CodeType 指定，
// 服务端首先使用 JSON 解码 Option，然后通过 Option 的 CodeType 解码剩余的内容。

//对于 GeeRPC 来说，目前需要协商的唯一一项内容是消息的编解码方式。我们将这部分信息，放到结构体 Option 中承载。
type Option struct {
	MagicNumber    int           // MagicNumber表示这是一个rpc请求
	CodecType      codec.Type    // 客户端可以选择不同的编解码器来编码主体
	ConnectTimeout time.Duration // 链接超时，0不设置
	HandleTimeout  time.Duration // 处理超时
}

// DefaultOption 默认信息
var DefaultOption = &Option{
	MagicNumber:    MagicNumber,
	CodecType:      codec.GobType,
	ConnectTimeout: time.Second * 10,
}

// Server rpc服务端
type Server struct {
	serviceMap sync.Map
}

// request 存储一个请求的信息
type request struct {
	h            *codec.Header // 请求头
	argv, replyv reflect.Value // 请求的argv和replyv
	mType        *methodType
	svc          *service
}

// NewServer 实例化
func NewServer() *Server {
	return &Server{}
}

// DefaultServer 默认的服务端实例
var DefaultServer = NewServer()

func (s *Server) Accept(lis net.Listener) {
	for { // for 循环等待 socket 连接建立
		conn, err := lis.Accept()
		if err != nil {
			log.Println("rpc server: accept error:", err)
			return
		}

		// 开启子协程处理，处理过程交给了 ServerConn 方法
		go s.ServeConn(conn)
	}
}

// ServeConn 的实现就和之前讨论的通信过程紧密相关了
// 首先使用 json.NewDecoder 反序列化得到 Option 实例，检查 MagicNumber 和 CodeType 的值是否正确
// 然后根据 CodeType 得到对应的消息编解码器，接下来的处理交给 serverCodec
func (s *Server) ServeConn(conn io.ReadWriteCloser) {
	defer func() { _ = conn.Close() }()
	var opt Option
	// json解析option
	if err := json.NewDecoder(conn).Decode(&opt); err != nil {
		log.Println("rpc server: option error: ", err)
		return
	}

	// 判断是否为rpc请求
	if opt.MagicNumber != MagicNumber {
		log.Printf("rpc server: invalid magic number %x", opt.MagicNumber)
		return
	}

	// 判断编码器是否正确
	f := codec.NewCodecFuncMap[opt.CodecType]
	if f == nil {
		log.Printf("rpc server: invalid codec type %s", opt.CodecType)
		return
	}

	s.serveCodec(f(conn), &opt)
}

// invalidRequest 是发生错误时响应argv的占位符
var invalidRequest = struct{}{}

// serveCodec 编码器处理请求
func (s *Server) serveCodec(cc codec.Codec, opt *Option) {
	sending := new(sync.Mutex) // 互斥锁确保发送完整的响应
	wg := new(sync.WaitGroup)  // 等待，直到所有的请求都得到处理
	// 在一次连接中，允许接收多个请求，即多个 request header 和 request body
	// 因此这里使用了 for 无限制地等待请求的到来，直到发生错误（例如连接被关闭，接收到的报文有问题等）
	// 这里需要注意的点有三个：
	// handleRequest 使用了协程并发执行请求。
	// 处理请求是并发的，但是回复请求的报文必须是逐个发送的，并发容易导致多个回复报文交织在一起，客户端无法解析。在这里使用锁(sending)保证
	// 尽力而为，只有在 header 解析失败时，才终止循环
	for {
		// 读取请求
		req, err := s.readRequest(cc)
		if err != nil {
			if req == nil {
				break // 不可能恢复，所以关闭连接
			}
			req.h.Error = err.Error()
			s.sendResponse(cc, req.h, invalidRequest, sending)
			continue
		}
		wg.Add(1)
		// 起一个协程处理请求
		go s.handleRequest(cc, req, sending, wg, opt.HandleTimeout)
	}
	wg.Wait() // 阻塞等待所有请求处理完毕
	_ = cc.Close()
}

// readRequest 读取请求
func (s *Server) readRequest(cc codec.Codec) (*request, error) {
	h, err := s.readRequestHeader(cc)
	if err != nil {
		return nil, err
	}
	req := &request{h: h}
	req.svc, req.mType, err = s.findService(h.ServiceMethod)
	if err != nil {
		return req, err
	}
	// 通过 newArgv() 和 newReplyv() 两个方法创建出两个入参实例
	req.argv = req.mType.newArgv()
	req.replyv = req.mType.newReplyv()

	argvi := req.argv.Interface()
	// argv 可能是值类型，也可能是指针类型，所以处理方式有点差异
	if req.argv.Type().Kind() != reflect.Ptr {
		argvi = req.argv.Addr().Interface()
	}

	if err = cc.ReadBody(argvi); err != nil {
		log.Println("rpc server: read argv error: ", err)
		return req, err
	}
	return req, nil
}

// readRequestHeader 读取请求头
func (s *Server) readRequestHeader(cc codec.Codec) (*codec.Header, error) {
	var h codec.Header
	// 读取请求头
	if err := cc.ReadHeader(&h); err != nil {
		if err != io.EOF && err != io.ErrUnexpectedEOF { // 如果不是读到结束符，则报错
			log.Println("rpc server: read header error:", err)
		}
		return nil, err
	}
	return &h, nil
}

// sendResponse 响应数据
func (s *Server) sendResponse(cc codec.Codec, h *codec.Header, body interface{}, sending *sync.Mutex) {
	sending.Lock()
	defer sending.Unlock()
	if err := cc.Write(h, body); err != nil {
		log.Println("rpc server: write response error: ", err)
	}
}

// handleRequest 处理请求
func (s *Server) handleRequest(cc codec.Codec, req *request, sending *sync.Mutex, wg *sync.WaitGroup, timeout time.Duration) {
	defer wg.Done()
	// 需要确保 sendResponse 仅调用一次，因此将整个过程拆分为 called 和 sent 两个阶段
	called := make(chan struct{})
	sent := make(chan struct{})
	go func() {
		// 通过 req.svc.call 完成方法调用，将 replyv 传递给 sendResponse 完成序列化
		err := req.svc.call(req.mType, req.argv, req.replyv)
		called <- struct{}{}
		if err != nil {
			req.h.Error = err.Error()
			s.sendResponse(cc, req.h, invalidRequest, sending)
			sent <- struct{}{}
			return
		}
		s.sendResponse(cc, req.h, req.replyv.Interface(), sending) // 响应
		sent <- struct{}{}
	}()

	if timeout == 0 {
		<-called
		<-sent
		return
	}

	select {
	case <-time.After(timeout):
		// time.After() 先于 called 接收到消息，说明处理已经超时，called 和 sent 都将被阻塞
		req.h.Error = fmt.Sprintf("rpc server: request handle timeout: expect within %s", timeout)
		s.sendResponse(cc, req.h, invalidRequest, sending)
	case <-called:
		<-sent
	}
}

// Register 服务注册
func (s *Server) Register(rcvr interface{}) error {
	service := newService(rcvr)
	if _, dup := s.serviceMap.LoadOrStore(service.name, service); dup {
		return errors.New("rpc: service already defined: " + service.name)
	}
	return nil
}

// Register 在DefaultServer中注册服务的方法。
func Register(rcvr interface{}) error {
	return DefaultServer.Register(rcvr)
}

// findService 服务发现
func (s *Server) findService(serviceMethod string) (svc *service, mType *methodType, err error) {
	dot := strings.LastIndex(serviceMethod, ".")
	if dot < 0 {
		err = errors.New("rpc server: service/method request ill-formed: " + serviceMethod)
		return
	}
	// 因为 ServiceMethod 的构成是 “Service.Method”，因此先将其分割成 2 部分，第一部分是 Service 的名称，第二部分即方法名。
	serviceName, methodName := serviceMethod[:dot], serviceMethod[dot+1:]
	// 在 serviceMap 中找到对应的 service 实例
	svci, ok := s.serviceMap.Load(serviceName)
	if !ok {
		err = errors.New("rpc server: can't find service " + serviceName)
		return
	}
	// 从 service 实例的 method 中，找到对应的 methodType
	svc = svci.(*service)
	mType = svc.method[methodName]
	if mType == nil {
		err = errors.New("rpc server: can't find method " + methodName)
	}
	return
}

// Accept 接收侦听器上的连接，并服务请求或每个传入的连接
func Accept(lis net.Listener) {
	DefaultServer.Accept(lis)
}

// ServeHTTP HTTP服务
func (s *Server) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	if req.Method != "CONNECT" {
		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
		w.WriteHeader(http.StatusMethodNotAllowed)
		_, _ = io.WriteString(w, "405 must CONNECT\n")
		return
	}

	conn, _, err := w.(http.Hijacker).Hijack()
	if err != nil {
		log.Print("rpc hijacking ", req.RemoteAddr, ": ", err.Error())
		return
	}
	_, _ = io.WriteString(conn, "HTTP/1.0 "+connected+"\n\n")
	s.ServeConn(conn)
}

// HandleHTTP 在rpcPath上为RPC消息注册一个HTTP处理程序
// 仍然需要调用http.Serve()，通常是在go语句中
func (s *Server) HandleHTTP() {
	http.Handle(defaultRPCPath, s)
	http.Handle(defaultDebugPath, debugHTTP{s})
	log.Println("rpc server debug path:", defaultDebugPath)
}

// HandleHTTP 默认服务器注册HTTP处理程序的一种方便的方法
func HandleHTTP() {
	DefaultServer.HandleHTTP()
}
